package com.wzk.dwetl

import java.util.Properties

import ch.hsr.geohash.GeoHash
import org.apache.spark.sql.SparkSession

object GeoHashDisGen {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName(this.getClass.getSimpleName)
      .master("local[*]")
      .getOrCreate()

    val pro = new Properties()
    pro.setProperty("user","root");
    pro.setProperty("password","root");
    val df = spark.read.jdbc("jdbc:mysql://linux01:3306/realtimedw", "t_genhash_dis", pro)

    df.createTempView("area")

    val genHash = (lat: Double,lng: Double) => {
      GeoHash.geoHashStringWithCharacterPrecision(lat,lng,5)
    }

    spark.udf.register("geohashDis",genHash)

    val res = spark.sql(
      """
        |select
        | geohashDis(BD09_LAT,BD09_LNG) as geo,
        | province,
        | city,
        | dist
        |from
        | area
        |
        |""".stripMargin)
    res.printSchema()
    res.show()

    res.write.parquet("hdfs://linux01:8020/geo/")
    spark.close()



  }
}
